package com.lilei.bigdata.hive.mr;

import com.lilei.bigdata.hive.mrbean.PageViewsBean;
import com.lilei.bigdata.hive.mrbean.VisitBean;
import com.lilei.bigdata.hive.mrbean.WebLogBean;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import javax.print.attribute.standard.NumberUp;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;

public class VisitLog {
    static class VisitLogMapper extends Mapper<LongWritable,Text,Text, PageViewsBean>{
        PageViewsBean v = new PageViewsBean();
        Text k = new Text();
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //以session 分发出去
            String line = value.toString();
            String[] fields = line.split("\001");
            int step =Integer.parseInt(fields[5]);
            v.set(fields[0], fields[1], fields[2], fields[3],fields[4], step, fields[6], fields[7], fields[8], fields[9]);
            k.set(v.getSession());
            context.write(k,v);
        }
    }

    static  class VisitLogReducer extends Reducer<Text, PageViewsBean,NullWritable, VisitBean>{
        @Override
        protected void reduce(Text session, Iterable<PageViewsBean> values, Context context) throws IOException, InterruptedException {
            ArrayList<PageViewsBean> pageViewsBeans = new ArrayList<>();

            for (PageViewsBean pageViewsBean:values){
                try {
                    PageViewsBean bean = new PageViewsBean();
                    BeanUtils.copyProperties(bean,pageViewsBean);
                    pageViewsBeans.add(bean);
                } catch (Exception e) {
                    e.printStackTrace();
                }

            }

            Collections.sort(pageViewsBeans, new Comparator<PageViewsBean>() {
                @Override
                public int compare(PageViewsBean o1, PageViewsBean o2) {
                    return o1.getStep() > o2.getStep() ? 1 : -1;
                }
            });

            VisitBean visitBean = new VisitBean();
            visitBean.setSession(session.toString());
            visitBean.setRemote_addr(pageViewsBeans.get(0).getRemote_addr());
            visitBean.setInTime(pageViewsBeans.get(0).getTimestr());
            visitBean.setOutTime(pageViewsBeans.get(pageViewsBeans.size() -1).getTimestr());
            visitBean.setInPage(pageViewsBeans.get(0).getRequest());
            visitBean.setOutPage(pageViewsBeans.get(pageViewsBeans.size() -1).getRequest());
            visitBean.setReferal(pageViewsBeans.get(0).getReferal());
            visitBean.setPageVisits(pageViewsBeans.size());
            context.write(NullWritable.get(),visitBean);
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        job.setJarByClass(VisitLog.class);

        job.setMapperClass(VisitLogMapper.class);
        job.setReducerClass(VisitLogReducer.class);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(PageViewsBean.class);

        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(VisitBean.class);

        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

//        FileInputFormat.setInputPaths(job, new Path("C:/weblog/output"));
//        FileOutputFormat.setOutputPath(job, new Path("c:/weblog/pageviews"));

        job.waitForCompletion(true);
    }
}
